/*
 * @Author: gonglf
 * @Date: 2022-08-22 15:27:14
 * @LastEditors: gonglf
 * @LastEditTime: 2022-08-31 14:58:28
 * @Description:
 *
 */
package main

import (
	"fmt"
	"io"
	"log"
	"net"

	"gitee.com/network-learning/rpc/7gRPC_oneStream/pb"
	"google.golang.org/grpc"
)

/*
1、基于gRPC和pubsub包，提供一个跨网络的发布和订阅系统
2、Publish是普通的RPC方法，Subscribe则是一个单向的流服务
3、因为Subscribe是服务端的单向流，因此生成的HelloService_SubscribeServer接口中只有Send方法
*/

type PubsubService struct {
	// pub  *pubsub.Publisher
	cMsg chan string
}

func NewPubsubService() *PubsubService {
	return &PubsubService{
		cMsg: make(chan string, 10),
	}
}

func (p *PubsubService) Publish(stream pb.PubsubService_PublishServer) error {
	fmt.Println("一个新的链接")
	for {
		arg, err := stream.Recv()
		if err != nil {
			if err == io.EOF {
				fmt.Println("断开了")
				return nil
			}
			fmt.Println("err", err)
			return err
		}
		msg := arg.GetValue()
		fmt.Println(msg)
		// p.pub.Publish(msg)
		// p.cMsg <- msg
	}

}

func (p *PubsubService) Subscribe(arg *pb.String, stream pb.PubsubService_SubscribeServer) error {

	for v := range p.cMsg {
		msg := v
		if err := stream.Send(&pb.String{Value: msg}); err != nil { //没有stream.RecvMsg()，没有接收方法
			return err
		}
	}

	return nil
}

func main() {
	grpcServer := grpc.NewServer()
	pb.RegisterPubsubServiceServer(grpcServer, NewPubsubService())

	lis, err := net.Listen("tcp", ":1234")
	if err != nil {
		log.Fatal(err)
	}
	grpcServer.Serve(lis)
}
